package net.bwie.lianxi.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class splitProcessFunction extends ProcessFunction<String, String> {

    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displaysTag;
    private final OutputTag<String> actionTag;

    public splitProcessFunction(OutputTag<String> errorTag, OutputTag<String> startTag,
                                OutputTag<String> displaysTag, OutputTag<String> actionTag) {
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displaysTag = displaysTag;
        this.actionTag = actionTag;
    }

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        //1.解析json
        JSONObject jsonObject = JSON.parseObject(value);

        //2.错误日志
        if (null != jsonObject.getString("err")){
            ctx.output(errorTag,value);
            jsonObject.remove("err");
        }

        //3.启动日志
        if (null!= jsonObject.getString("start")){
            ctx.output(startTag,value);
            jsonObject.remove("start");
        }
        //4.页面日志
        else {
            JSONArray displays = jsonObject.getJSONArray("displays");
            JSONArray actions = jsonObject.getJSONArray("actions");

            jsonObject.remove("displays");
            jsonObject.remove("actions");

            //输出页面日志
            out.collect(jsonObject.toJSONString());

            //5.曝光日志
            if (null != displays && !displays.isEmpty()){
                for (int i = 0; i < displays.size(); i++) {
                    JSONObject displayJsonObject = displays.getJSONObject(i);
                    // 添加字段值
                    jsonObject.put("display", displayJsonObject);
                    // 输出
                    ctx.output(displaysTag, jsonObject.toJSONString());
                }
            }

            jsonObject.remove("display");

            //6.动作日志
            if (null != actions && !actions.isEmpty()){
                for (int i = 0; i < actions.size(); i++) {
                    JSONObject actionJsonObject = actions.getJSONObject(i);
                    // 添加字段值
                    jsonObject.put("action", actionJsonObject);
                    // 输出
                    ctx.output(actionTag, jsonObject.toJSONString());
                }
            }
        }
    }
}
